package org.medusa.db.handlersocket;

import java.util.HashMap;
import java.util.Map;
import java.util.Vector;

import org.medusa.db.IDatabase;
import org.medusa.log.ILog;
import org.medusa.network.tcp.TcpSyncLineConnector;
import org.medusa.packet.blob.LinePacket;

import com.sun.corba.se.impl.orbutil.concurrent.Mutex;

//ps: DTC含义：String dtc = dbName + ":" + tableName + ":" + cols;

//增强：使用过程中连接断掉后 缓存指令到某个地方，再次连接上后按照顺序同步到数据库中
//增强2：集群、双向同步、主从模式、主主模式,
//集群中要判断同ip多次添加，要禁止掉，返回失败
//断线重连失败后立即使用另一个连接/ 重连N次后使用集群中其他数据库
//3.插入、删除、更新 后将完整记录返回 (nosql没有使用＊遍历所有字段的能力，未必可行）
//4.断线重连：最后一次使用的index需要在认证信息发送后重新请求一次 (bug: 虽然使用了重新打开index的机制，但是op命令所用的pid已经无效了
//5. index -> pid/indexid/iid/idxid

public class HandlerSocket1 implements IDatabase {

	// info
	private String name;
	private ILog log;

	// auth
	private String host_r;
	private String host_wr;
	private int port_r;
	private int port_wr;
	private String port_passwd_r;
	private String port_passwd_wr;

	// connector
	private TcpSyncLineConnector conn_r;
	private TcpSyncLineConnector conn_wr;

	// thread_safe
	private boolean thread_safe_r;
	private boolean thread_safe_wr;
	private Mutex mutex_r;
	private Mutex mutex_wr;

	// index
	int index_max;
	private int pid_counter_r;
	private int pid_counter_wr;
	private Map<Integer, String> pid_dtc_link_r;
	private Map<String, Integer> dtc_pid_link_r;
	private Map<Integer, String[]> pid_columns_link; // pid-查询的列
	private Map<Integer, String> pid_dtc_link_wr;
	private Map<String, Integer> dtc_pid_link_wr;

	// private Integer last_op_pid_r;
	// private Integer last_op_pid_wr;

	public HandlerSocket1(int index_max, String name, ILog log,
			boolean thread_safe_r, boolean thread_safe_wr) {
		this.name = "HandlerSocket:" + name;
		this.log = log;

		// this.auth = new HandlerSocketAuth();
		this.host_r = "localhost";
		this.host_wr = "localhost";
		this.port_r = 9998;
		this.port_wr = 9999;
		this.port_passwd_r = "";
		this.port_passwd_wr = "";

		this.conn_r = new TcpSyncLineConnector(1024000, 1024);
		this.conn_wr = new TcpSyncLineConnector(1024000, 1024);

		this.thread_safe_r = thread_safe_r;
		this.thread_safe_wr = thread_safe_wr;
		this.mutex_r = new Mutex();
		this.mutex_wr = new Mutex();

		this.index_max = index_max;
		this.pid_counter_r = 0;
		this.pid_counter_wr = 0;
		this.pid_dtc_link_r = new HashMap<Integer, String>();
		this.dtc_pid_link_r = new HashMap<String, Integer>();
		this.pid_columns_link = new HashMap<Integer, String[]>();
		this.pid_dtc_link_wr = new HashMap<Integer, String>();
		this.dtc_pid_link_wr = new HashMap<String, Integer>();

		// this.last_op_pid_r = -1;
		// this.last_op_pid_wr = -1;

	}

	public HandlerSocket1(String name, ILog log, boolean thread_safe_r,
			boolean thread_safe_wr) {
		this(Integer.MAX_VALUE, name, log, thread_safe_r, thread_safe_wr);
	}

	public Error Connect(String host_r, int port_r, String host_wr, int port_wr) {
		Error err = null;
		/*
		 * if (this.auth == null) { this.auth = new HandlerSocketAuth(); }
		 */

		// If already connected return
		if (this.conn_r.IsConnected()) {
			err = new Error("Already connected to server(r);");
			this.Error(err);
			return err;
		} else {
			this.host_r = host_r;
			this.port_r = port_r;
			err = this.connect_r();
		}

		if (this.conn_wr.IsConnected()) {
			err = new Error("Already connected to server(wr);");
			this.Error(err);
		} else {
			this.host_wr = host_wr;
			this.port_wr = port_wr;
			err = this.connect_wr();
		}
		return err;
	}

	public Error Reconnect() {

		Error err = this.reconnect_r();
		if (err == null) {
			err = this.reconnect_wr();
		}

		return err;
	}

	public Error Auth(String port_passwd_r, String port_passwd_wr) {

		this.port_passwd_r = port_passwd_r;
		this.port_passwd_wr = port_passwd_wr;

		Error err = this.auth_r();
		if (err == null) {
			err = this.auth_wr();
		}

		return err;
	}

	public InsertResult Insert(String dbName, String tableName,
			String indexName, String[] columns, Object[] vals) {
		this.Debug("Insert called");
		InsertResult result = new InsertResult();
		this.lock_wr();
		// ------------------------
		IndexResult ires = this.open_index_wr(dbName, tableName, indexName,
				this.join_string(columns, ","));

		if (ires.Error() == null) {
			String cols = this.join_hs_field(vals); // strings.Join(vals, "\t");
			// String strindex = ires.Index().toString();
			String col_count = "" + vals.length;
			String oper = "+";
			String[] args = new String[] { oper, col_count, cols };

			// this.lock_wr()
			// ------------------------
			CommandResponse response = this.hs_insert_command(ires.PID(), args);

			if (response.result_code == 1) {
				result.Set("INSERT: Data Exists:" + response.data[1], -2);
				this.Warning(result.ErrorMessage());
			} else if (response.result_code != 0) {
				result.Set("Error Inserting Data:" + response.data[1], -1);
				this.Error(result.ErrorMessage());
			} else {
				if (response.data.length >= 2) {// 存在自增id
					result.IncrementID(Integer.parseInt(response.data[1].trim()));
					this.Debug("Insert return, last insert id:",
							result.IncrementID()); // LAST_INSERT_ID
					// Increment
					// id
				} else { // 不存在自增id
					result.IncrementID(0);
					this.Debug("Insert return, have no increment field.");
				}
			}
		} else {
			result.Error(ires.ErrorMessage());
		}
		// ------------------------
		this.unlock_wr();
		this.Debug("Insert finish, err:", result.ErrorMessage());
		return result;

	}

	public ModifyResult Modify(String dbName, String tableName,
			String indexName, String oper, Object[] cmpvals, String[] columns,
			Object[] newvals) {

		ModifyResult result = this.Modify(dbName, tableName, indexName, oper,
				cmpvals, columns, newvals, Integer.MAX_VALUE, 0);
		return result;
	}

	public ModifyResult Modify(String dbName, String tableName,
			String indexName, String oper, Object[] cmpvals, String[] columns,
			Object[] newvals, int limit, int skip) {

		this.Debug("Modify called");
		ModifyResult result = new ModifyResult();

		this.lock_wr();
		// ------------------------
		IndexResult ires = this.open_index_wr(dbName, tableName, indexName,
				this.join_string(columns, ","));
		if (ires.Error() == null) {
			String query = join_hs_field(cmpvals); // strings.Join(keys, "\t")
			String queryCount = "" + cmpvals.length;

			String[] args = new String[] { oper, queryCount, query };

			// this.lock_wr()
			// ------------------------
			CommandResponse response = this.hs_modify_command(ires.PID(), args,
					limit, skip, "U", newvals);

			if (response.result_code == 1) {
				result.Set("Error Something:" + response.data[1].trim(), 0);
				this.Error(result.ErrorMessage());
			} else {
				result.AffectedRows(Integer.parseInt(response.data[1].trim()));
				this.Debug("Modify return, affected rows:",
						result.AffectedRows());
			}
		} else {
			result.Error(ires.ErrorMessage());
		}
		// ------------------------
		this.unlock_wr();
		this.Debug("Modify finish, err:", result.ErrorMessage());
		return result;
	}

	public ModifyResult Remove(String dbName, String tableName,
			String indexName, String oper, Object[] cmpvals) {
		ModifyResult result = this.Remove(dbName, tableName, indexName, oper,
				cmpvals, Integer.MAX_VALUE, 0);
		return result;
	}

	public ModifyResult Remove(String dbName, String tableName,
			String indexName, String oper, Object[] cmpvals, int limit, int skip) {

		this.Debug("Remove called");
		ModifyResult result = new ModifyResult();

		this.lock_wr();
		// ------------------------
		IndexResult ires = this.open_index_wr(dbName, tableName, indexName, "");
		if (ires.Error() == null) {
			String query = join_hs_field(cmpvals); // strings.Join(keys, "\t")
			String queryCount = "" + cmpvals.length;

			String[] args = new String[] { oper, queryCount, query };

			// this.lock_wr()
			// ------------------------
			CommandResponse response = this.hs_modify_command(ires.PID(), args,
					limit, skip, "D");

			if (response.result_code == 1) {
				result.Set("Error Something:" + response.data[1].trim(), 0);
				this.Error(result.ErrorMessage());
			} else {
				result.AffectedRows(Integer.parseInt(response.data[1].trim()));
				this.Debug("Remove return, affected rows:",
						result.AffectedRows());
			}
		} else {
			result.Error(ires.ErrorMessage());
		}
		// ------------------------
		this.unlock_wr();
		this.Debug("Remove finish, err:", result.ErrorMessage());
		return result;

	}

	public QueryResult Query(String dbName, String tableName, String indexName,
			String oper, Object[] keys, String[] columns, int limit, int skip) {

		this.Debug("Query called");

		QueryResult result = new QueryResult();

		this.lock_r();
		// ------------------------
		IndexResult ires = this.open_index_r(dbName, tableName, indexName,
				this.join_string(columns, ","));

		if (ires.Error() == null) {
			String cols = join_hs_field(keys); // strings.Join(vals, "\t");
			// String strpid = ires.PID().toString();
			String col_count = "" + keys.length;
			String[] args = new String[] { oper, col_count, cols };

			// this.lock_r()
			// ------------------------
			CommandResponse response = this.hs_query_command(ires.PID(), args,
					limit, skip);
			if (response.result_code != 0) {
				result.Set("Error Something:" + response.data[1].trim(), null);
				this.Error(result.ErrorMessage());
			} else {
				result.Rows(this.parseQueryResult(ires.PID(), response));
				this.Debug("Query returned:", result.Rows().size(), "rows");
				if (this.log.GetLevel() >= ILog.DebugLv) {
					String info = "\n";
					info += this.join_string(columns, "\t");
					info += "\n--------------------------------------------------------------------------\n";

					for (int i = 0; i < result.Rows().size(); i++) {
						HandlerSocketRow row = result.Rows().get(i);
						String[] row_field = new String[columns.length];// make([]string,
																		// len(columns));
						for (int idx = 0; idx < columns.length; idx++) {
							row_field[idx] = row.GetField_String(columns[idx]);
						}
						// fmt.Println(strings.Join(row_field, "\t| "))
						info += this.join_string(row_field, "\t") + "\n";
					}
					// info +=
					// "--------------------------------------------------------------------------\n";
					// info+="\n";
					this.Debug(info);
				}
			}

		} else {
			result.Error(ires.ErrorMessage());
		}
		// ------------------------
		this.unlock_r();
		this.Debug("Query finish, err:", result.ErrorMessage());
		return result;
	}

	public QueryResult Query(String dbName, String tableName, String indexName,
			String oper, Object[] keys, String[] columns) {
		return this.Query(dbName, tableName, indexName, oper, keys, columns,
				Integer.MAX_VALUE, 0);
	}

	public QueryOneResult QueryOne(String dbName, String tableName,
			String indexName, String oper, Object[] keys, String[] columns) {

		QueryOneResult result = new QueryOneResult();
		this.Debug("QueryOne called");

		this.lock_r();
		// ------------------------
		IndexResult ires = this.open_index_r(dbName, tableName, indexName,
				this.join_string(columns, ","));

		if (ires.Error() == null) {
			String cols = this.join_hs_field(keys); // strings.Join(vals, "\t");
			// String strpid = ires.PID().toString();
			String col_count = "" + keys.length;
			String[] args = new String[] { oper, col_count, cols };

			// this.lock_r()
			// ------------------------
			CommandResponse response = this.hs_query_command(ires.PID(), args,
					1, 0);
			if (response.result_code != 0) {
				result.Set("Error Something:" + response.data[1].trim(), null);
				this.Error(result.ErrorMessage());
			} else {
				Vector<HandlerSocketRow> rows = this.parseQueryResult(
						ires.PID(), response);
				if (rows.size() >= 1) {
					result.Error(null);
					result.Row(rows.get(0));
				}
				this.Debug("QueryOne returned:", rows.size(), " rows");
				if (this.log.GetLevel() >= ILog.DebugLv) {
					String info = "\n";
					info += this.join_string(columns, "\t");
					info += "\n--------------------------------------------------------------------------\n";
					if (result.Row().NotNull()) {
						String[] row_field = new String[columns.length];// make([]string,
																		// len(columns));
						for (int idx = 0; idx < columns.length; idx++) {
							row_field[idx] = result.Row().GetField_String(
									columns[idx]);
						}
						// fmt.Println(strings.Join(row_field, "\t"));
						info += this.join_string(row_field, "\t");
					}
					info += "\n";
					this.Debug(info);
				}
			}

		} else {
			result.Error(ires.ErrorMessage());
		}
		// ------------------------
		this.unlock_r();
		this.Debug("QueryOne finish, err:", result.ErrorMessage());
		return result;

	}

	public Error Close() {
		this.close_r();
		this.close_wr();
		return null;
	}

	public void Error(Object... args) {
		if (this.log != null)
			this.log.Error(this.name, args);
	}

	public void Warning(Object... args) {
		if (this.log != null)
			this.log.Warning(this.name, args);
	}

	public void Success(Object... args) {
		if (this.log != null)
			this.log.Success(this.name, args);
	}

	public void Debug(Object... args) {
		if (this.log != null)
			this.log.Debug(this.name, args);
	}

	public void Stack(Object... args) {
		if (this.log != null)
			this.log.Stack(this.name, args);
	}

	public void Temp(Object... args) {
		if (this.log != null)
			this.log.Temp(this.name, args);
	}

	private void lock_r() {
		if (this.thread_safe_r) {
			try {
				this.mutex_r.acquire();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private void unlock_r() {
		if (this.thread_safe_r) {
			this.mutex_r.release();
		}
	}

	private void lock_wr() {
		if (this.thread_safe_wr) {
			try {
				this.mutex_wr.acquire();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	private void unlock_wr() {
		if (this.thread_safe_wr) {
			this.mutex_wr.release();
		}
	}

	private Error close_r() {

		this.Debug("Close(r) called");
		Error err = null;// new Error("succeed!");
		String errinfo = "";
		// If not connected return
		if (!this.conn_r.IsConnected()) {
			errinfo += "A connection to a MySQL server is required to use this function(r)";
		} else {

			this.Debug("Sent quit command to server(r)");

			// Close connection(r)
			this.conn_r.Close();
			this.Debug("Closed connection to server(r)");

			// Free indexes
			this.pid_dtc_link_r.clear();
			this.dtc_pid_link_r.clear();
			this.pid_columns_link.clear();
			this.pid_counter_r = 0;
		}

		if (/*errinfo != ""*/!"".equals(errinfo)) {
			err = new Error(errinfo);
			this.Error(errinfo);
		}
		this.Debug("Close(r) finish, err:", err);
		return err;
	}

	private Error close_wr() {

		this.Debug("Close(wr) called");
		Error err = null;
		String errinfo = "";
		if (!this.conn_wr.IsConnected()) {
			errinfo += "A connection to a MySQL server is required to use this function(wr);";
		} else {

			this.Debug("Sent quit command to server(wr)");

			// Close connection(wr)
			this.conn_wr.Close();
			this.Debug("Closed connection to server(wr)");

			// Free indexes
			// this.pid_dtc_link_wr.clear();
			// this.dtc_pid_link_wr.clear();
			// this.pid_counter_wr = 0;
		}

		if (/*errinfo != ""*/!"".equals(errinfo)) {
			err = new Error(errinfo);
			this.Debug(errinfo);
		}
		this.Debug("Close(wr) finish, err:", err);
		return err;
	}

	private Error connect_r() {
		this.Debug("Connect(r) called, host:", this.host_r, " port:",
				this.host_r);
		Error err = this.conn_r.Connect(this.host_r, this.port_r, 1000);
		if (err == null) {
			this.pid_dtc_link_r.clear();
			this.dtc_pid_link_r.clear();
			this.pid_columns_link.clear();
			this.pid_counter_r = 0;
			// this.last_op_time_r = new Date().getTime();
			this.Success("Connect(r) succeed!");
		} else {
			err = new Error("Error Connect(r):" + err.getMessage());
			this.Error(err);
		}
		return err;
	}

	private Error connect_wr() {
		this.Debug("Connect(wr) called, host:", this.host_wr, " port:",
				this.host_wr);
		Error err = this.conn_wr.Connect(this.host_wr, this.port_wr, 1000);
		if (err == null) {
			this.pid_dtc_link_wr.clear();
			this.dtc_pid_link_wr.clear();
			this.pid_counter_wr = 0;
			// this.last_op_time_wr = new Date().getTime();
			this.Success("Connect(wr) succeed!");
		} else {
			err = new Error("Error Connect(wr):" + err.getMessage());
			this.Error(err);
		}
		return err;
	}

	private Error reconnect_r() {
		this.Debug("Reconnect(r) called");
		this.close_r();
		this.connect_r();
		this.auth_r();
		// if (this.last_op_pid_r >= 0)
		// this.open_index_r(this.pid_dtc_link_r.get(this.last_op_pid_r));

		return null;
	}

	private Error reconnect_wr() {
		this.Debug("Reconnect(wr) called");
		this.close_wr();
		this.connect_wr();
		this.auth_wr();
		// if (this.last_op_pid_wr >= 0)
		// this.open_index_wr(this.pid_dtc_link_wr.get(this.last_op_pid_wr));

		return null;
	}

	private Error auth_r() {
		Error err = null;
		if (!this.conn_r.IsConnected()) {
			err = new Error("have no connected to mysql(r)!");
			this.Error(err);
			return err;
		}

		if ("".equals(port_passwd_r)/*this.port_passwd_r == ""*/) {
			return err;
		}

		this.Debug("Auth(r) called, passwd:", this.port_passwd_r);

		String col_count = "1";

		String[] args_r = new String[] { col_count, this.port_passwd_r };
		CommandResponse message_r = this.hs_auth_command(this.conn_r, args_r);

		if (message_r.result_code != 0) {
			err = new Error("Error Auth(r):" + message_r.data[1].trim());
			this.Error(err);
		} else {
			this.Success("Auth(r) succeed!");
		}

		return err;
	}

	private Error auth_wr() {
		Error err = null;
		if (!this.conn_wr.IsConnected()) {
			err = new Error("have no connected to mysql(wr)!");
			this.Error(err);
			return err;
		}
		if ("".equals(port_passwd_wr)/*this.port_passwd_wr == ""*/) {
			return err;
		}

		this.Debug("Auth(wr) called, passwd:", this.port_passwd_wr);

		String col_count = "1";

		String[] args_wr = new String[] { col_count, this.port_passwd_wr };
		CommandResponse message_wr = this
				.hs_auth_command(this.conn_wr, args_wr);
		if (message_wr.result_code != 0) {
			err = new Error("Error Auth(wr):"
					+ message_wr.data[1].toLowerCase());
			this.Error(err);
		} else {
			this.Success("Auth(wr) succeed!");
		}

		return err;
	}

	/*
	 * private IndexResult open_index_r(String dtc) { String[] dtc_arr =
	 * dtc.split("$"); return this .open_index_r(dtc_arr[0], dtc_arr[1],
	 * dtc_arr[2], dtc_arr[3]); }
	 */

	private IndexResult open_index_r(String dbName, String tableName,
			String indexName, String columns) {

		// String cols = this.join_string(columns, ",");// columns
		String dtc = dbName + "$" + tableName + "$" + indexName + "$" + columns;
		IndexResult result = new IndexResult(dtc);

		this.check_timeout_r();

		if (!this.conn_r.IsConnected()) {
			result.Error("have no connected to mysql(r)!");
			this.Error(result.ErrorMessage());
			return result;
		}

		// this.lock_index_r()
		// ------------------------
		if (!this.dtc_pid_link_r.containsKey(dtc)) {

			// pid limit
			if (this.pid_counter_r >= this.index_max) {
				result.PID(1);
			} else {
				result.PID(this.pid_counter_r);
			}

			this.Debug("open index(r) called: ", result.PID(), "\t", dtc);

			// String strpid = ires.PID().toString();
			String[] args = new String[] { result.PID().toString(), dbName,
					tableName, indexName, columns };

			CommandResponse response = this.hs_open_command(this.conn_r, args);

			if (response.result_code != 0) {
				result.Error("Error Opening Index(r):"
						+ response.data[1].trim());
				this.Error("Error Opening Index(r):" + result.ErrorMessage());
			} else {
				// Update indexes
				this.pid_counter_r += 1;

				this.pid_dtc_link_r.put(result.PID(), dtc);
				this.dtc_pid_link_r.put(dtc, result.PID());
				this.pid_columns_link.put(result.PID(), columns.split(","));

				this.Debug("create index(r) succeed: ", result.PID(), "\t", dtc);
			}
		} else
			result.PID(this.dtc_pid_link_r.get(dtc));
		// ------------------------
		// this.unlock_index_r()

		return result;
	}

	/*
	 * private IndexResult open_index_wr(String dtc) { String[] dtc_arr =
	 * dtc.split("$"); this.open_index_wr(dtc_arr[0], dtc_arr[1], dtc_arr[2],
	 * dtc_arr[3]); return null; }
	 */

	private IndexResult open_index_wr(String dbName, String tableName,
			String indexName, String columns) {
		// String cols = this.join_string(columns, ",");// columns
		String dtc = dbName + "$" + tableName + "$" + indexName + "$" + columns;
		IndexResult result = new IndexResult(dtc);

		this.check_timeout_wr();

		if (!this.conn_wr.IsConnected()) {
			result.Error("have no connected to mysql(wr)!");
			this.Error(result.ErrorMessage());
			return result;
		}

		// this.lock_index_wr()
		// ------------------------
		if (!this.dtc_pid_link_wr.containsKey(dtc)) {

			// pid limit
			if (this.pid_counter_wr >= this.index_max) {
				result.PID(1);
			} else {
				result.PID(this.pid_counter_wr);
			}

			this.Debug("open index(wr) called: ", result.PID(), "\t", dtc);

			// String strpid = ires.PID().toString();
			String[] args = new String[] { result.PID().toString(), dbName,
					tableName, indexName, columns };
			CommandResponse response = this.hs_open_command(this.conn_wr, args);

			if (response.result_code != 0) {
				result.Error("Error Opening Index(wr)"
						+ response.data[1].trim());
				this.Error("Error Opening Index(wr):" + result.ErrorMessage());
			} else {
				// Update indexes
				this.pid_counter_wr += 1;

				this.pid_dtc_link_wr.put(result.PID(), dtc);
				this.dtc_pid_link_wr.put(dtc, result.PID());
				this.Debug("create index(wr) succeed: ", result.PID(), "\t",
						dtc);
			}
		} else
			result.PID(this.dtc_pid_link_wr.get(dtc));
		// ------------------------
		// this.unlock_index_wr()

		return result;
	}

	private Vector<HandlerSocketRow> parseQueryResult(int pid,
			CommandResponse hsp) {
		Vector<HandlerSocketRow> rows = new Vector<HandlerSocketRow>();
		int fieldCount = Integer.parseInt(hsp.data[0]);
		int remainingFields = hsp.data.length - 1;
		if (fieldCount > 0) {
			int rs = remainingFields / fieldCount;
			// rows = new HandlerSocketRow[rs];

			int offset = 1;

			for (int row_idx = 0; row_idx < rs; row_idx++) {
				Map<String, String> row_data = new HashMap<String, String>();// make(map[string]string,
				// fieldCount)
				for (int fld_idx = 0; fld_idx < fieldCount; fld_idx++) {

					// 在这里转义效率太低，因为很多字段都不是string的，还会破坏非string字段的完整性
					// ff := strings.Replace(hsp.data[offset+fld_idx], TMTab,
					// Tab, -1)
					// ff = strings.Replace(ff, TMReturn, Return, -1)
					// row_data[this.query_indexes[pid][fld_idx]] = ff

					row_data.put(this.pid_columns_link.get(pid)[fld_idx],
							hsp.data[offset + fld_idx]);
				}
				rows.add(new HandlerSocketRow(row_data));// [row_idx] = ;
				offset += fieldCount;
			}
		}
		return rows;
	}

	private CommandResponse hs_auth_command(TcpSyncLineConnector conn,
			String[] params) {
		String command = "A";

		String cmd = String.format("%s\t%s\n", command,
				this.join_string(params, "\t"));
		this.Stack("auth cmd:[" + cmd + "]");

		return this.run_command(conn, cmd);
	}

	private CommandResponse hs_open_command(TcpSyncLineConnector conn,
			String[] params) {
		String command = "P";
		String cmd = String.format("%s\t%s\n", command,
				this.join_string(params, "\t"));

		this.Stack("open cmd:[" + cmd + "]");

		return this.run_command(conn, cmd);
	}

	private CommandResponse hs_insert_command(int pid_wr, String[] params) {
		// this.last_op_pid_wr = pid_wr;
		String cmd = String.format("%d\t%s\n", pid_wr,
				this.join_string(params, "\t"));
		this.Stack("insert cmd:[" + cmd + "]");

		return this.run_command(this.conn_wr, cmd);
	}

	private CommandResponse hs_modify_command(int pid_wr, String[] criteria,
			int limit, int skip, String mop, Object[] newvals) {
		// String mop = "U";
		// this.last_op_pid_wr = pid_wr;
		String cmd = String.format("%d\t%s\t%d\t%d\t%s\t%s\n", pid_wr,
				join_string(criteria, "\t"), limit, skip, mop,
				this.join_hs_field(newvals));
		this.Stack("modify cmd:[" + cmd + "]");

		return this.run_command(this.conn_wr, cmd);
	}

	private CommandResponse hs_modify_command(int pid_wr, String[] criteria,
			int limit, int skip, String mop) {
		// this.last_op_pid_wr = pid_wr;
		return this.hs_modify_command(pid_wr, criteria, limit, skip, mop,
				new Object[] {});
	}

	private CommandResponse hs_query_command(int pid_r, String[] params,
			int limit, int skip) {
		// this.last_op_pid_r = pid_r;
		String cmd = String.format("%d\t%s\t%d\t%d\n", pid_r,
				this.join_string(params, "\t"), limit, skip);
		this.Stack("query cmd:[" + cmd + "]");

		return this.run_command(this.conn_r, cmd);
	}

	/*
	 * private void open_index_by_info_r(int pid, String dtc) {
	 * 
	 * }
	 */

	private CommandResponse run_command(TcpSyncLineConnector conn, String cmd) {
		LinePacket sent = new LinePacket(4096);// Michael
		sent.WriteBytes(cmd);

		
		LinePacket recv = conn.SendCommand(sent);

		while (recv == null) {
			try {
				Thread.sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			if (conn == this.conn_r) {
				this.reconnect_r();
				// 重新打开所需的pid
				// this.open_index_by_info_r(pid, dtc);
			} else if (conn == this.conn_wr) {
				this.reconnect_wr();
				// 重新打开所需的pid
				// this.open_index_by_info_wr(pid, dtc);
			}
			recv = conn.SendCommand(sent);
		}

		// clean last op pid
		// if (conn == this.conn_r)
		// this.last_op_pid_r = -1;
		// else if (conn == this.conn_wr)
		// this.last_op_pid_wr = -1;
		CommandResponse response = new CommandResponse(recv);

		return response;
	}

	private void check_timeout_r() {
		if (!this.conn_r.IsConnected()) {
			this.reconnect_r();
		}
	}

	private void check_timeout_wr() {
		if (!this.conn_wr.IsConnected()) {
			this.reconnect_wr();
		}
	}

	private String join_string(String[] columns, String flag) {
		String result = "";
		if (columns != null) {
			for (int i = 0; i < columns.length; i++) {
				if (i != columns.length - 1)
					result += columns[i] + flag;
				else
					result += columns[i];
			}
		}

		return result;
	}

	private String join_hs_field(Object... vals) {
		String result = "";
		for (int i = 0; i < vals.length; i++) {
			if (i != vals.length - 1)
				result += vals[i] + "\t";
			else
				result += vals[i];
		}

		return result;
	}

}
